package com.lagou.kafka.demo.producer;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * @param
 * @author oyz
 * @version 1.0
 * @description:  生产者1
 * @date 2022/3/25 15:05
 */
public class Myproducer2 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String,Object> configs = new HashMap<>(3);
        // 连接的broker的地址
        configs.put("bootstrap.servers","192.168.106.130:9092");
        // 指定key的序列化
        configs.put("key.serializer", IntegerSerializer.class);
        // 指定value的序列化
        configs.put("value.serializer", StringSerializer.class);
        //configs.put("acks","all");
        //configs.put("reties","3");

        // 用于设置用户自定义的消息头
        List<Header> headers = new ArrayList<>();
        headers.add(new RecordHeader("biz.name","producer.demo".getBytes()));
        for (int i = 0; i < 100 ; i++) {

            ProducerRecord<Integer,String> record = new ProducerRecord<>("topic_1",0,i,"helle lagou"+i,headers);

            KafkaProducer<Integer,String> producer = new KafkaProducer<Integer, String>(configs);


            // 消息异步的确认方式
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if(exception == null) {
                        System.out.println("消息的主题："+metadata.topic());
                        System.out.println("消息的分区号："+metadata.partition());
                        System.out.println("消息的偏移量："+metadata.offset());
                    }else{
                        System.out.println("错误的异常信息:"+exception.getMessage());
                    }
                }
            });
        }





    }
}
